package tableAndSql;

import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.HashMap;

/**
 * @desc
 * @Author xuyouchang
 * @Date 2022/2/24
 */
public class TwoStreamJoin {
    public static void main(String[] args) {
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.enableCheckpointing(10000L);
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage("hdfs://172.17.1.102:8020/checkpoint/");
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(streamExecutionEnvironment);


        streamTableEnvironment.executeSql("create table `user` (`id` INT, `name` STRING, `age` INT, `time` TIMESTAMP(3)) " +
                "with ('connector' = 'kafka','topic' = 'user','properties.bootstrap.servers' = " +
                "'172.17.1.101:9092','properties.group.id' = 'test6Group','scan.startup.mode' = 'earliest-offset','format' = 'json')");

        streamTableEnvironment.executeSql("create table exam (`user_id` INT, `score` INT, `time` TIMESTAMP(3)) " +
                "with ('connector' = 'kafka','topic' = 'exam','properties.bootstrap.servers' = " +
                "'172.17.1.101:9092','properties.group.id' = 'test6Group','scan.startup.mode' = 'earliest-offset','format' = 'json')");

        streamTableEnvironment.executeSql("create table res (`id` INT, `name` STRING, `age` INT,`score` INT, PRIMARY KEY (id) NOT ENFORCED) " +
                "with ('connector' = 'elasticsearch-7','index' = 'merge_xyc','hosts' = " +
                "'172.17.1.101:9200')");

        Table table = streamTableEnvironment.sqlQuery("select u.id, u.name, u.age, e.score from `user` u FULL OUTER join exam e on (u.id = e.user_id " +
                "and e.`time` between u.`time` - INTERVAL '4' MINUTE and u.`time` + INTERVAL '5' MINUTE)");
        table.executeInsert("res");

    }
}
